/*
 * Copyright (C) 2008 Universidade Federal de Campina Grande
 *  
 * This file is part of OurGrid. 
 *
 * OurGrid is free software: you can redistribute it and/or modify it under the
 * terms of the GNU Lesser General Public License as published by the Free 
 * Software Foundation, either version 3 of the License, or (at your option) 
 * any later version. 
 * 
 * This program is distributed in the hope that it will be useful, but WITHOUT 
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or 
 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License
 * for more details. 
 * 
 * You should have received a copy of the GNU Lesser General Public License 
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 * 
 */
package org.ourgrid.peer.controller.allocation;

import java.util.List;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import org.ourgrid.common.interfaces.LocalWorkerProvider;
import org.ourgrid.common.interfaces.RemoteWorkerProvider;
import org.ourgrid.common.interfaces.management.RemoteWorkerManagement;
import org.ourgrid.common.interfaces.management.WorkerManagement;
import org.ourgrid.common.interfaces.to.LocalWorkerState;
import org.ourgrid.common.interfaces.to.RequestSpec;
import org.ourgrid.common.spec.worker.WorkerSpec;
import org.ourgrid.common.statistics.control.WorkerControl;
import org.ourgrid.peer.PeerConfiguration;
import org.ourgrid.peer.PeerConstants;
import org.ourgrid.peer.controller.WorkerNotificationController;
import org.ourgrid.peer.controller.messages.RequestMessages;
import org.ourgrid.peer.controller.messages.WorkerMessages;
import org.ourgrid.peer.dao.AllocationDAO;
import org.ourgrid.peer.dao.RequestDAO;
import org.ourgrid.peer.to.AllocableWorker;
import org.ourgrid.peer.to.LocalAllocableWorker;
import org.ourgrid.peer.to.LocalConsumer;
import org.ourgrid.peer.to.LocalWorker;
import org.ourgrid.peer.to.RemoteAllocableWorker;
import org.ourgrid.peer.to.Request;
import org.ourgrid.reqtrace.Req;

import br.edu.ufcg.lsd.commune.container.servicemanager.ServiceManager;
import br.edu.ufcg.lsd.commune.identification.DeploymentID;
import br.edu.ufcg.lsd.commune.identification.ServiceID;
import br.edu.ufcg.lsd.commune.network.certification.CertificationUtils;

public class RedistributionController {

	private static RedistributionController instance = null;
	
	private RedistributionController() {}
	
	public static RedistributionController getInstance() {
		if (instance == null) {
			instance = new RedistributionController();
		}
		return instance;
	}
	
	/**
	 * Redistributes a local worker
	 * @param allocableWorker 
	 * @param localWorker
	 * @return true if the local worker was allocated, false if it's idle
	 */
	public void redistributeDeliveredWorker(LocalAllocableWorker allocableWorker, LocalWorker localWorker, ServiceManager serviceManager) {
		Request suitableRequestForWorker = DefaultAllocator.getInstance().getRequestForWorkerSpec(
				localWorker.getWorkerSpec(), serviceManager);
		
		if(suitableRequestForWorker == null){
			allocableWorker.getWorkerManagement().stopWorking();
			return;
		}
		
		allocateRequestToIdleWorker(allocableWorker, suitableRequestForWorker, localWorker, serviceManager);
	}

	public boolean redistributeIdleWorker(LocalWorker localWorker, ServiceManager serviceManager) {
		Request suitableRequestForWorker = DefaultAllocator.getInstance().getRequestForWorkerSpec(
				localWorker.getWorkerSpec(), serviceManager);
		
		if(suitableRequestForWorker != null){
			AllocationDAO dao = serviceManager.getDAO(AllocationDAO.class);
			AllocableWorker allocableWorker = dao.getAllocableWorker(localWorker.getPublicKey());
			allocateRequestToIdleWorker(allocableWorker, suitableRequestForWorker, localWorker, serviceManager);
			return true;
		}
		
		return false;
	}

	/**
	 * Redistributes a remote worker
	 * @param provider
	 * @param worker
	 * @param workerSpec
	 * @return true if the local worker was allocated, false if it was 
	 * disposed to its provider
	 */
	public boolean redistributeWorker(RemoteWorkerProvider provider, RemoteAllocableWorker allocable,
			ServiceManager serviceManager) {
		
		DeploymentID workerDeploymentID = serviceManager.getStubDeploymentID(allocable.getWorkerManagement());
		
		if (workerDeploymentID == null || !isThereAtLeastOneRequestThatMatchesAndNeedMoreWorkers(
				serviceManager.getDAO(RequestDAO.class).getRunningRequests(), allocable.getWorkerSpec())) {
			
			DeploymentID stubDeploymentID = serviceManager.getStubDeploymentID(provider);
			boolean providerIsUp = stubDeploymentID != null;

			if (providerIsUp) {
				serviceManager.getLog().debug(WorkerMessages.getDisposingWorkerToRemoteProviderMessage(
						stubDeploymentID, workerDeploymentID));
				
				disposeWorkerToRemoteProvider(allocable, serviceManager);
				serviceManager.release(allocable.getWorkerManagement());
			}
			return false;
		}

		Request request = DefaultAllocator.getInstance().getRequestForWorkerSpec(allocable.getWorkerSpec(), serviceManager);
		
		RemoteAllocableWorker rAlloca = 
			serviceManager.getDAO(AllocationDAO.class).getRemoteAllocableWorker(workerDeploymentID.getPublicKey());
		
		LocalConsumer localConsumer = request.getConsumer();
		rAlloca.setConsumer(localConsumer);
		rAlloca.setRequest(request);
        request.addAllocableWorker(rAlloca);
		
		rAlloca.workForMyGrid(request.getConsumer().getDeploymentID());

		if(!request.needMoreWorkers()) {
			Future<?> future = serviceManager.getDAO(RequestDAO.class).removeFuture(request.getSpec());
			if (future != null) {
				future.cancel(true);
			}
		}
		
		return true;
	}

	/**
	 * Redistributes an AllocableWorker
	 * @param allocable
	 */
	@Req({"REQ015", "REQ016"})
	public void redistributeWorker(AllocableWorker allocable, ServiceManager serviceManager) {
		
		boolean workerIsUp = serviceManager.getStubDeploymentID(allocable.getWorkerManagement()) != null;
		RequestSpec requestSpec = allocable.getRequest().getSpec();
		
		allocable.deallocate();
		
		if(allocable.isWorkerLocal()){
			
			LocalWorker localWorker = ((LocalAllocableWorker) allocable).getLocalWorker();

			if (workerIsUp) {
				redistributeLocalWorker(localWorker, (LocalAllocableWorker) allocable, serviceManager);
				
			} else {
				WorkerNotificationController.workerFailure(serviceManager, allocable.getWorkerSpec().getServiceID(), 
						localWorker);
			}
			
			
		} else {

			RemoteAllocableWorker rAllocable = (RemoteAllocableWorker) allocable;
			if (workerIsUp) {
				redistributeRemoteWorker(rAllocable, serviceManager);
				
			} else {
				disposeWorkerToRemoteProvider(rAllocable, serviceManager);
			}
		}
		
		Request request = serviceManager.getDAO(RequestDAO.class).getRequest(requestSpec.getRequestId());
		if (request != null && request.needMoreWorkers()) {
			scheduleRequest(requestSpec, serviceManager);
		}
	}
	
	private void disposeWorkerToRemoteProvider(RemoteAllocableWorker rAllocable, ServiceManager serviceManager) {
		
		ServiceID workerServiceID = rAllocable.getServiceID();
		RemoteWorkerProvider provider = rAllocable.getWorkerProvider();

		WorkerControl.getInstance().removeRemoteWorker(workerServiceID, serviceManager);
		
		serviceManager.getDAO(AllocationDAO.class).removeRemoteAllocableWorker(workerServiceID.getPublicKey());

		DeploymentID stubDeploymentID = serviceManager.getStubDeploymentID(provider);
		boolean providerIsUp = stubDeploymentID != null;

		if (providerIsUp) {
			provider.disposeWorker(workerServiceID);
			
			RemoteWorkerManagement workerManagement = rAllocable.getWorkerManagement();
			boolean workerIsUp = serviceManager.getStubDeploymentID(workerManagement) != null;
			if (workerIsUp) {
				serviceManager.release(workerManagement);
			}
		}
	}
	
	private boolean isThereAtLeastOneRequestThatMatchesAndNeedMoreWorkers(
			List<Request> requests, WorkerSpec spec) {
		
		for (Request request : requests) {
			if (Util.matchAndNeedWorkers(spec, request)) {
				return true;
			}
		}
		
		return false;
	}
	
	@Req("REQ117")
	public void createAllocableWorker(LocalWorker localWorker, ServiceManager serviceManager) {
		
		AllocationDAO dao = serviceManager.getDAO(AllocationDAO.class);
		
		LocalWorkerProvider localWorkerProvider = (LocalWorkerProvider) serviceManager.getObjectDeployment(
				PeerConstants.LOCAL_ACCESS_OBJECT_NAME).getObject();
		
		String myCertPathDN = CertificationUtils.getCertSubjectDN(serviceManager.getMyCertPath());
		
		LocalAllocableWorker allocableWorker = new LocalAllocableWorker(localWorker,
				localWorkerProvider,myCertPathDN);
		
		dao.addLocalAllocableWorker(localWorker.getPublicKey(), allocableWorker);
	}
	
	@Req("REQ117")
	private void allocateRequestToIdleWorker(AllocableWorker allocableWorker, Request request, LocalWorker localWorker, ServiceManager serviceManager) {
		
		localWorker.setStatus(LocalWorkerState.IN_USE);
		
		WorkerControl.getInstance().statusChanged(
				localWorker.getServiceID(), LocalWorkerState.IN_USE, 
				request.getConsumer().getDeploymentID().getContainerID().getUserAtServer(),
				serviceManager);
		
		LocalConsumer localConsumer = request.getConsumer();
		allocableWorker.setConsumer(localConsumer);
		allocableWorker.setRequest(request);
	    request.addAllocableWorker(allocableWorker);
	    allocableWorker.workForMyGrid(request.getConsumer().getDeploymentID());
	    
	    if(!request.needMoreWorkers()){
			Future<?> future = serviceManager.getDAO(RequestDAO.class).removeFuture(request.getSpec());
			if (future != null) {
				future.cancel(true);
			}
	    }
	}
	
	
	/**
	 * @param requestSpec
	 */
	private void scheduleRequest(RequestSpec requestSpec, ServiceManager serviceManager) {
		
		RequestDAO requestDAO = serviceManager.getDAO(RequestDAO.class);
		
		if(! requestDAO.containsAScheduledRequest(requestSpec)) {
			int delay = Integer.parseInt(serviceManager.getContainerContext().getProperty(
					PeerConfiguration.PROP_REPEAT_REQUEST_DELAY));
			
			Future<?> future = serviceManager.scheduleActionWithFixedDelay(PeerConstants.REQUEST_WORKERS_ACTION_NAME, delay, 
					TimeUnit.SECONDS, requestSpec.getRequestId());
			
			requestDAO.putScheduledRequest(requestSpec, future);
			serviceManager.getLog().debug(RequestMessages.getRequestRepetitionMessage(delay, requestSpec.getRequestId()));
		}
	}
	
	@Req("REQ015")
	private void redistributeRemoteWorker(RemoteAllocableWorker allocableWorker, ServiceManager serviceManager) {

		RemoteWorkerProvider remoteWorkerProvider = (RemoteWorkerProvider) allocableWorker.getWorkerProvider();
		
		
		redistributeWorker(remoteWorkerProvider, allocableWorker, serviceManager);
	}

	@Req("REQ015")
	private void redistributeLocalWorker(LocalWorker localWorker, LocalAllocableWorker allocableWorker, 
			ServiceManager serviceManager) {
		
		redistributeDeliveredWorker(allocableWorker, localWorker, serviceManager);
		
		serviceManager.registerInterest(PeerConstants.WORKER_MONITOR_OBJECT_NAME, 
				allocableWorker.getWorkerID().toString(), WorkerManagement.class);
	}
	
}
